package com.reload.stream.flow;

import java.io.IOException;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @Date: 2024/12/4 13:55
 * @Author: jianchenlong
 * @Description:
 */
public class FlowDemo {
    public static void main(String[] args) throws IOException {
        // (原生方式)定义一个发布者 发布数据
        Flow.Publisher<String> publisher = new Flow.Publisher<String>() {
            private Flow.Subscriber<? super String> subscriber;

            // 订阅者会订阅这个发布者接口
            @Override
            public void subscribe(Flow.Subscriber<? super String> subscriber) {
                this.subscriber = subscriber;
            }
        };

        // 提交一个数据的发布者
        SubmissionPublisher<String> submissionPublisher = new SubmissionPublisher<String>();

        // 定义一个订阅者订阅数据
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {

            private Flow.Subscription subscription;

            // 在订阅时  onXXXX 事件触发机制。 在xxxx事件发生的时候执行这个回调
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread().getName() + "-订阅开始了-" + subscription);
                this.subscription = subscription;
                // 请求1个数据
                this.subscription.request(1);
            }

            // 在下一个元素到达时，接受到了新数据
            @Override
            public void onNext(String item) {
                System.out.println(Thread.currentThread().getName() + "-订阅者接收到新数据-" + item);
                this.subscription.request(1);
            }

            // 在错误发生时
            @Override
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread().getName() + "-订阅者接收到错误-" + throwable);
            }

            // 在完成时
            @Override
            public void onComplete() {
                System.out.println(Thread.currentThread().getName() + "-生产者完成");

            }
        };
        // 绑定发布者和订阅者
        submissionPublisher.subscribe(subscriber);
        // 发布者有数据，订阅者就会拿到
        for (int i = 0; i < 10; i++) {
            // Publisher发布的所有数据在它的buffer区
            // 发布者发布10条数据
            submissionPublisher.submit("submission - " + i);
            // 发布者对数据进行消费
            // submissionPublisher.consume(item -> {
            //     System.out.println("发送了一个item = " + item);
            // });
        }
        System.in.read();
    }
}
